package com.atguigu.flink.day08;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author Felix
 * @date 2023/12/9
 * 该案例演示了算子状态-ListState
 * 需求：在map算子中计算每个并行度(分区)数据的个数。
 */
public class Flink07_OpeState_ListState {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        env
            .socketTextStream("hadoop102", 8888)
            .map(new MyCountMapFunction())
            .print();

        env.execute();

    }

    private static class MyCountMapFunction implements MapFunction<String,String> , CheckpointedFunction {
        private int count = 0;
        private ListState<Integer> countState;
        @Override
        public String map(String value) throws Exception {
            ++count;
            return "个数：" + count;
        }

        @Override
        public void snapshotState(FunctionSnapshotContext context) throws Exception {
            System.out.println("snapshotState...");
            //先清状态
            countState.clear();
            //将本地变量的值  放到状态中保存起来
            countState.add(count);
        }

        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {
            System.out.println("~~~initializeState~~~");
            //当初始化方法执行的时候，判断是从快照中(检查点)恢复状态还是新建窗台
            //因为我们现在没有开启检查点，所以不会对状态进行备份，所以isRestored返回值是false

            ListStateDescriptor<Integer> listStateDescriptor
                = new ListStateDescriptor<Integer>("countState", Integer.class);
            countState = context.getOperatorStateStore().getListState(listStateDescriptor);

            if(context.isRestored()){
                //从状态中获取数据
                Integer next = countState.get().iterator().next();
                //将从状态中拿到的值 赋给本地变量
                count = next;
            }
        }
    }
}
